package org.example.tinyurl.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.hash.Hashing;
import org.apache.dubbo.config.annotation.DubboService;
import org.example.tinyurl.api.TinyUrlService;
import org.example.tinyurl.entity.ResultStatus;
import org.example.tinyurl.entity.RpcResult;
import org.example.tinyurl.entity.TblUrl;
import org.example.tinyurl.mapper.TinyUrlMapper;
import org.example.util.ContextHolderUtil;
import org.example.util.DateUtils;
import org.example.util.NumberUtil;
import org.redisson.api.RBloomFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.web.client.RestTemplate;

import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.concurrent.TimeUnit;

/**
 * @author: zk
 * @date: 2023/6/19 10:40
 */
@DubboService
public class TinyUrlServiceImpl extends ServiceImpl<TinyUrlMapper, TblUrl> implements TinyUrlService {

    private static final int RETRY_COUNT = 3;
    private static final String PREFIX = "abc";
    private static final String SUFFIX = "xyz";
    private static final String TINY_LONG_URL = "tiny_long_url:";
    private static final int CACHE_EXPIRE_DAYS = 30;
    private static final int DB_EXPIRE_DAYS = 365;
    private static final String GET_ID_URL = "http://127.0.0.1:8080/api/segment/get/leaf-segment-test";

    private static final Logger logger = LoggerFactory.getLogger(TinyUrlServiceImpl.class);

    @Autowired
    private TinyUrlMapper tinyUrlMapper;

    @Autowired
    private RBloomFilter bloomFilter;

    @Autowired
    private RedisTemplate redisTemplate;

    private RestTemplate restTemplate = new RestTemplate();

    /**
     * 方案3：BloomFilter -> Redis -> Mysql
     *
     * @param longUrl
     * @return
     */
    @Override
    public RpcResult<String> generateTinyUrl(String longUrl) {
        ValueOperations opsForValue = redisTemplate.opsForValue();
        String tempLongUrl = longUrl;
        Long hashCode;
        String tinyUrl;

        for (int i = 0; i < RETRY_COUNT; i++) {
            hashCode = getMurmurhashCode(tempLongUrl);
            tinyUrl = NumberUtil.base10ToBase62(hashCode);
            String redisKey = TINY_LONG_URL + tinyUrl;
            if (bloomFilter.contains(tinyUrl)) {
                String longUrlCache = (String) opsForValue.get(redisKey);
                if (longUrlCache != null) {
                    if (longUrlCache.equals(longUrl)) {
                        // Redis续期
                        redisTemplate.expire(redisKey, CACHE_EXPIRE_DAYS, TimeUnit.DAYS);
                        return new RpcResult<>(true, ResultStatus.OK, tinyUrl);
                    }
                    // 短链接存在，长链接不相等，说明冲突
                    logger.warn("{} 已存在，正在进行第 {} 次重试......", tinyUrl, i + 1);
                    tempLongUrl = PREFIX + tempLongUrl + SUFFIX;
                    continue;
                }
            }

            // BloomFilter中没有短链接，或者Redis中短链接过期或不存在，再查MySQL
            ContextHolderUtil.idHolder.set(hashCode);
            // 设置记录信息
            TblUrl recode = new TblUrl();
            recode.setTinyUrl(tinyUrl);
            recode.setLongUrl(longUrl);
            recode.setCreateTime(DateUtils.getCurrentDate());
            try {
                tinyUrlMapper.insert(recode);
            } catch (DuplicateKeyException e) {
                logger.warn("{} 已存在，正在进行第 {} 次重试......", tinyUrl, i + 1);
                tempLongUrl = PREFIX + tempLongUrl + SUFFIX;
                continue;
            }

            // 插入成功
            bloomFilter.add(tinyUrl);
            opsForValue.set(redisKey, longUrl, CACHE_EXPIRE_DAYS, TimeUnit.DAYS);
            return new RpcResult<>(true, ResultStatus.OK, tinyUrl);
        }
        return new RpcResult<>(false, ResultStatus.DUPLICATE_ERROR, null);
    }

    /**
     * 由短链接得到长链接
     *
     * @param tinyUrl
     * @return
     */
    @Override
    public RpcResult<String> getLongUrl(String tinyUrl) {
        ValueOperations opsForValue = redisTemplate.opsForValue();
        String redisKey = TINY_LONG_URL + tinyUrl;
        // 1. 先经过布隆过滤器
        if (bloomFilter.contains(tinyUrl)) {
            // 2. 再经过Redis
            String longUrlCache = (String) opsForValue.get(redisKey);
            if (longUrlCache != null) {
                // Redis中有短链接：续期并返回长链接
                redisTemplate.expire(redisKey, CACHE_EXPIRE_DAYS, TimeUnit.DAYS);
                return new RpcResult<>(true, ResultStatus.OK, longUrlCache);
            }

            // 3. Redis中短链接过期或不存在，再查MySQL
            LambdaQueryWrapper<TblUrl> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(TblUrl::getTinyUrl, tinyUrl);
            TblUrl tblUrl = tinyUrlMapper.selectOne(queryWrapper);
            if (tblUrl != null) {
                // 判断是否过期
                Timestamp createTs = DateUtils.date2Timestamp(tblUrl.getCreateTime());
                Timestamp expireTs = DateUtils.getTimeAfterDay(createTs, DB_EXPIRE_DAYS);
                Timestamp currentTs = DateUtils.getCurrentTime();
                if (expireTs.before(currentTs)) {
                    logger.error("短链接 {} 已过期", tinyUrl);
                    return new RpcResult<>(false, ResultStatus.EXPIRE, null);
                }

                String longUrl = tblUrl.getLongUrl();
                if (logger.isDebugEnabled()) {
                    logger.debug("所在数据库下标 : {}", (tblUrl.getId() % 4 + 1));
                }
                opsForValue.set(redisKey, longUrl, CACHE_EXPIRE_DAYS, TimeUnit.DAYS);
                tblUrl.setCreateTime(currentTs);
                tinyUrlMapper.updateById(tblUrl);
                return new RpcResult<>(true, ResultStatus.OK, longUrl);
            }
        }
        logger.warn("{} 短链接不存在！", tinyUrl);
        return new RpcResult<>(false, ResultStatus.NOT_EXIST_ERROR, null);
    }

    /**
     * 计算 hash 值
     *
     * @param str
     * @return
     */
    public static Long getMurmurhashCode(String str) {
        return Math.abs(Hashing.murmur3_128().hashString(str, StandardCharsets.UTF_8).asLong());
    }
}
